/****************************************************************************
 *
 * Copyright (C) 2003-2008 Los Alamos National Security, LLC
 *                         Packet Analytics Corporation
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License Version 2 as
 * published by the Free Software Foundation.  You may not use, modify or
 * distribute this program under any other version of the GNU General
 * Public License.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 *
 ****************************************************************************/
package nfse.stats;

import java.io.*;
import java.net.*;
import java.util.concurrent.Semaphore;

import nfse.NetFSE;
import nfse.NetFSERecord;
import nfse.NetFSEUtil;
import nfse.SensorInfo;

public class StatsSocket extends Thread {
    private Socket sock;

    private PrintWriter out = null;

    private File f = null;

    private final Semaphore available = new Semaphore(1, true);

    private StatsServer statsServer;

    public StatsSocket(Socket s, StatsServer statsServer) {
        sock = s;
        this.statsServer = statsServer;

    }

    public void writeToFile(String statsEntry) throws Exception {
        available.acquire();
        out.println(statsEntry);
        available.release();
    }

    public void rotateFile() throws Exception {
        available.acquire();
        if (out != null) {
            out.flush();
            out.close();
        }
        f = File.createTempFile("stats", ".csv", new File(NetFSE.TEMP_DIR));
        out = new PrintWriter(new FileWriter(f));
        statsServer.addStatsFile(f);
        available.release();
    }

    public void run() {
        try {
            long startMS = System.currentTimeMillis();
            DataInputStream dis = new DataInputStream(sock.getInputStream());
            DataOutputStream dos = new DataOutputStream(sock.getOutputStream());
            int sensorID = dis.readInt(); // read the sensor ID

            SensorInfo sensorInfo = NetFSE.getSensorInfo(sensorID);

            NetFSERecord rec = ((NetFSERecord) NetFSE.TYPE_CLASSES.get(new Integer(sensorInfo.getToolID())));

            rotateFile();

            long ts = System.currentTimeMillis(); // Unix time in MS used to
            // monitor throughput
            int len = 0;
            long count = 0;
            try {
                while (true) {
                    try {
                        len = dis.readInt();
                    } catch (Exception e) {
                        len = -1;
                    }
                    if (len == -1) {
                        break;
                    }
                    int bytesRead = 0;
                    byte[] dataBytes = new byte[len];
                    while (bytesRead < len)
                        bytesRead += dis.read(dataBytes, bytesRead, len - bytesRead);
                    if (bytesRead != len) {
                        System.err.println("Bad data, not enough bytes.");
                    } else {

                        writeToFile(new String(dataBytes));

                        count++;

                        if (count % 50000 == 0) {
                            rotateFile();
                            System.out.println("Rotated: '" + f.getAbsolutePath() + "'");
                        }

                        if (count % 10000 == 0) {
                            // Output statistics periodically
                            long ts2 = System.currentTimeMillis();
                            long ms = ts2 - ts;
                            long elapsedMS = ts2 - startMS;
                            float elapsedSec = (float) elapsedMS / (float) 1000;
                            float sec = (float) ms / (float) 1000;
                            float sps1 = (float) count / elapsedSec;
                            float sps2 = (float) 10000 / sec;
                            System.out.println("StatsServer: Total Records=" + NetFSEUtil.formattedInteger(count)

                            + "; Current RPS= " + NetFSEUtil.formattedDouble(sps2) + "; Average RPS= "
                                    + NetFSEUtil.formattedDouble(sps1) + "");
                            ts = System.currentTimeMillis();
                        }
                    }
                }

            } catch (Exception E) {
                E.printStackTrace();
            }
            out.flush();
            out.close();
            dis.close();
            dos.close();
            sock.close();
            System.out.println("Stats socket exited.");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}